package drds.binlog.server;

import com.google.common.base.Function;
import com.google.common.collect.ComputingMap;
import drds.binlog.binlog_task.BinlogTask;
import drds.binlog.binlog_task.BinlogTaskFactory;
import drds.binlog.binlog_task.IBinlogTask;
import drds.binlog.common.AbstractLifeCycle;
import drds.binlog.common.Authors;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.position.LogPosition;
import drds.binlog.common.position.Position;
import drds.binlog.common.position.PositionRange;
import drds.binlog.store.Event;
import drds.binlog.store.Events;
import drds.binlog.store.IEventStore;
import drds.common.$;
import drds.common.Author;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Slf4j
@Author(name = Authors.LI_YANG)
public class Server extends AbstractLifeCycle implements IServer, IService
{
    public static final String destination_key = "destination";
    public static final String binlog_task_key = BinlogTask.class.getSimpleName();
    @Setter
    @Getter
    private Map<String, IBinlogTask> destinationToBinlogTaskMap;
    @Setter
    @Getter
    private BinlogTaskFactory binlogTaskFactory;


    public Server()
    {

    }


    public void start()
    {
        if (!isRunning())
        {
            super.start();


            destinationToBinlogTaskMap = ComputingMap.makeComputingMap(new Function<String, IBinlogTask>()
            {

                public IBinlogTask apply(String destination)
                {
                    return binlogTaskFactory.config(destination, "");
                }
            });
        }
    }

    public void stop()
    {
        super.stop();
        for (Map.Entry<String, IBinlogTask> entry : destinationToBinlogTaskMap.entrySet())
        {
            try
            {
                IBinlogTask binlogTask = entry.getValue();
                if (binlogTask.isRunning())
                {
                    try
                    {
                        String destination = entry.getKey();
                        MDC.put(destination_key, destination);
                        entry.getValue().stop();
                        log.info("stop " + binlog_task_key + "[{}] successfully", destination);
                    } finally
                    {
                        MDC.remove(destination_key);
                    }
                }
            } catch (Exception e)
            {
                log.error(String.format("stop " + binlog_task_key + "[%s] has an error", entry.getKey()), e);
            }
        }

    }

    //
    public void start(final String destination)
    {
        final IBinlogTask binlogTask = destinationToBinlogTaskMap.get(destination);
        if (!binlogTask.isRunning())
        {
            try
            {
                MDC.put(destination_key, destination);
                binlogTask.start();
                log.info("start " + binlog_task_key + "[{}] successfully", destination);
            } finally
            {
                MDC.remove(destination_key);
            }
        }
    }

    public void stop(String destination)
    {
        IBinlogTask binlogTask = destinationToBinlogTaskMap.remove(destination);
        if (binlogTask != null)
        {
            if (binlogTask.isRunning())
            {
                try
                {
                    MDC.put(destination_key, destination);
                    binlogTask.stop();
                    log.info("stop " + binlog_task_key + " [{}] successfully", destination);
                } finally
                {
                    MDC.remove(destination_key);
                }
            }
        }
    }
    //


    /**
     * 客户端订阅，重复订阅时会更新对应的filter信息
     */
    @Override
    public void subscribe(ClientInfo clientInfo)
    {
        checkStart(clientInfo.getDestination());

        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(clientInfo.getDestination());
        if (!binlogTask.getMetaDataManager().isRunning())
        {
            binlogTask.getMetaDataManager().start();
        }

        binlogTask.getMetaDataManager().subscribe(clientInfo); // 执行一下meta订阅

        Position position = binlogTask.getMetaDataManager().getPosition(clientInfo);
        if (position == null)
        {
            position = binlogTask.getEventStore().getFirstPosition();// 获取一下store中的第一条
            if (position != null)
            {
                binlogTask.getMetaDataManager().updatePosition(clientInfo, position); // 更新一下cursor
            }
            log.info("subscribe successfully, {} with first :{} ", clientInfo, position);
        } else
        {
            log.info("subscribe successfully, use last cursor :{} ", clientInfo, position);
        }

        // 通知下订阅关系变化
        binlogTask.subscribeChange(clientInfo);
    }

    /**
     * 取消订阅
     */
    @Override
    public void unsubscribe(ClientInfo clientInfo)
    {
        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(clientInfo.getDestination());
        binlogTask.getMetaDataManager().unsubscribe(clientInfo); // 执行一下meta订阅
        log.info("unsubscribe successfully, {}", clientInfo);
    }

    //
    public boolean isStart(String destination)
    {
        return destinationToBinlogTaskMap.containsKey(destination) && destinationToBinlogTaskMap.get(destination).isRunning();
    }

    /**
     * 查询所有的订阅信息
     */
    public List<ClientInfo> getClientInfoList(String destination)
    {
        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(destination);
        return binlogTask.getMetaDataManager().getClientInfoList(destination);
    }
    //

    /**
     * 获取数据
     * <pre>
     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
     * </pre>
     */
    @Override
    public Message getWithAutoAck(ClientInfo clientInfo, int batchSize)
    {
        return getWithAutoAck(clientInfo, batchSize, null, null);
    }

    /**
     * 获取数据，可以指定超时时间.
     * <p>
     * <pre>
     * 几种case:
     * a. 如果timeout为null，则采用tryGet方式，即时获取
     * b. 如果timeout不为null
     *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
     *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
     *
     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
     * </pre>
     */
    @Override
    public Message getWithAutoAck(ClientInfo clientInfo, int batchSize, Long timeout, TimeUnit timeUnit)

    {
        checkStart(clientInfo.getDestination());
        checkHasSubscribed(clientInfo);
        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(clientInfo.getDestination());
        synchronized (binlogTask)
        {
            // 获取到流式数据中的最后一批获取的位置
            PositionRange<LogPosition> lastestPositionRange = binlogTask.getMetaDataManager().getLastestPositionRange(clientInfo);
            if (lastestPositionRange != null)
            {
                throw new IllegalStateException(String.format("clientId:%s has last batch:[%s] isn't ack , maybe loss data", clientInfo.getClientId(), lastestPositionRange));
            }
            //
            Position startPosition = binlogTask.getMetaDataManager().getPosition(clientInfo);
            Events<Event> events = getEvents(binlogTask.getEventStore(), startPosition, batchSize, timeout, timeUnit);
            if (!$.isHasData(events.getEventList()))
            {
                log.debug("getWithAutoAck successfully, clientId:{} BatchSize:{} but result is null", clientInfo.getClientId(), batchSize);
                return new Message(-1, new ArrayList()); // 返回空包，避免生成batchId，浪费性能
            } else
            {
                // 记录到流式信息
                Long batchId = binlogTask.getMetaDataManager().addPositionRange(clientInfo, events.getPositionRange());

                // 直接提交ack
                ack(clientInfo, batchId);//获取之后立即ack
                return new Message(batchId, null);//byteStringList
            }
        }
    }

    /**
     * 不指定 setIndexAsOriginIndexAndOffset$originIndex 获取事件。canal 会记住此 client 最新的 setIndexAsOriginIndexAndOffset$originIndex。 <br/>
     * 如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。
     * <p>
     * <pre>
     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
     * </pre>
     */
    @Override
    public Message getWithoutAck(ClientInfo clientInfo, int batchSize)
    {
        return getWithoutAck(clientInfo, batchSize, null, null);
    }

    /**
     * 不指定 setIndexAsOriginIndexAndOffset$originIndex 获取事件。canal 会记住此 client 最新的 setIndexAsOriginIndexAndOffset$originIndex。 <br/>
     * 如果是第一次 fetch，则会从 canal 中保存的最老一条数据开始输出。
     * <p>
     * <pre>
     * 几种case:
     * a. 如果timeout为null，则采用tryGet方式，即时获取
     * b. 如果timeout不为null
     *    1. timeout为0，则采用get阻塞方式，获取数据，不设置超时，直到有足够的batchSize数据才返回
     *    2. timeout不为0，则采用get+timeout方式，获取数据，超时还没有batchSize足够的数据，有多少返回多少
     *
     * 注意： meta获取和数据的获取需要保证顺序性，优先拿到meta的，一定也会是优先拿到数据，所以需要加同步. (不能出现先拿到meta，拿到第二批数据，这样就会导致数据顺序性出现问题)
     * </pre>
     */
    @Override
    public Message getWithoutAck(ClientInfo clientInfo, int batchSize, Long timeout, TimeUnit timeUnit)

    {
        checkStart(clientInfo.getDestination());
        checkHasSubscribed(clientInfo);

        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(clientInfo.getDestination());
        synchronized (binlogTask)
        {
            // 获取到流式数据中的最后一批获取的位置
            PositionRange<LogPosition> lastestPositionRange = binlogTask.getMetaDataManager().getLastestPositionRange(clientInfo);

            Events<Event> events = null;
            if (lastestPositionRange != null)
            { // 存在流数据
                events = getEvents(binlogTask.getEventStore(), lastestPositionRange.getStartPosition(), batchSize, timeout, timeUnit);
            } else
            {// ack后第一次获取
                Position startPosition = binlogTask.getMetaDataManager().getPosition(clientInfo);
                if (startPosition == null)
                { // 第一次，还没有过ack记录，则获取当前store中的第一条
                    startPosition = binlogTask.getEventStore().getFirstPosition();
                }
                events = getEvents(binlogTask.getEventStore(), startPosition, batchSize, timeout, timeUnit);
            }

            if (!$.isHasData(events.getEventList()))
            {
                return new Message(-1, new ArrayList()); // 返回空包，避免生成batchId，浪费性能
            } else
            {
                // 记录到流式信息
                Long batchId = binlogTask.getMetaDataManager().addPositionRange(clientInfo, events.getPositionRange());
                events.getEventList();

                return new Message(batchId, null);//byteStringList
            }

        }
    }


    /**
     * 进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
     * <p>
     * <pre>
     * 注意：进行反馈时必须按照batchId的顺序进行ack(需有客户端保证)
     * </pre>
     */
    @Override
    public void ack(ClientInfo clientInfo, long batchId)
    {
        checkStart(clientInfo.getDestination());
        checkHasSubscribed(clientInfo);

        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(clientInfo.getDestination());
        PositionRange<LogPosition> positionRange = binlogTask.getMetaDataManager().removePositionRange(clientInfo, batchId); // 更新位置
        if (positionRange == null)
        { // 说明是重复的ack/rollback
            throw new IllegalStateException(String.format("ack error , clientId:%s batchId:%d is not exist , please check",
                    clientInfo.getClientId(),
                    batchId));
        }

        // 更新cursor
        if (positionRange.getAckPosition() != null)
        {
            binlogTask.getMetaDataManager().updatePosition(clientInfo, positionRange.getAckPosition());
        }
        // 可定时清理数据
        binlogTask.getEventStore().ack(positionRange.getEndPosition());

    }

    /**
     * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
     */
    @Override
    public void rollback(ClientInfo clientInfo)
    {
        checkStart(clientInfo.getDestination());
        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(clientInfo.getDestination());
        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = binlogTask.getMetaDataManager().hasSubscribed(clientInfo);
        if (!hasSubscribe)
        {
            return;
        }

        synchronized (binlogTask)
        {
            // 清除batch信息
            binlogTask.getMetaDataManager().clearBatchIdToPositionRangeMap(clientInfo);
            // rollback eventStore中的状态信息
            binlogTask.getEventStore().rollback();
            log.info("rollback successfully, clientId:{}", new Object[]{clientInfo.getClientId()});
        }
    }

    /**
     * 回滚到未进行 {@link #ack} 的地方，下次fetch的时候，可以从最后一个没有 {@link #ack} 的地方开始拿
     */
    @Override
    public void rollback(ClientInfo clientInfo, Long batchId)
    {
        checkStart(clientInfo.getDestination());
        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(clientInfo.getDestination());

        // 因为存在第一次链接时自动rollback的情况，所以需要忽略未订阅
        boolean hasSubscribe = binlogTask.getMetaDataManager().hasSubscribed(clientInfo);
        if (!hasSubscribe)
        {
            return;
        }
        synchronized (binlogTask)
        {
            // 清除batch信息
            PositionRange<LogPosition> positionRange = binlogTask.getMetaDataManager().removePositionRange(clientInfo,
                    batchId);
            if (positionRange == null)
            { // 说明是重复的ack/rollback
                throw new IllegalStateException(String.format("rollback error, clientId:%s batchId:%d is not exist , please check",
                        clientInfo.getClientId(),
                        batchId));
            }

            binlogTask.getEventStore().rollback();// rollback
            // eventStore中的状态信息
            log.info("rollback successfully, clientId:{} batchId:{} setIndexAsOriginIndexAndOffset$originIndex:{}",
                    clientInfo.getClientId(),
                    batchId,
                    positionRange);
        }
    }


    /**
     * 根据不同的参数，选择不同的方式获取数据
     */
    private Events<Event> getEvents(IEventStore IEventStoreStore, Position startPosition, int batchSize, Long timeout,
                                    TimeUnit timeUnit)
    {
        if (timeout == null)
        {
            return IEventStoreStore.tryGet(startPosition, batchSize);
        } else
        {
            try
            {
                if (timeout <= 0)
                {
                    return IEventStoreStore.get(startPosition, batchSize);
                } else
                {
                    return IEventStoreStore.get(startPosition, batchSize, timeout, timeUnit);
                }
            } catch (Exception e)
            {
                throw new IllegalStateException(e);
            }
        }
    }


    private void checkHasSubscribed(ClientInfo clientInfo)
    {
        IBinlogTask binlogTask = destinationToBinlogTaskMap.get(clientInfo.getDestination());
        boolean hasSubscribe = binlogTask.getMetaDataManager().hasSubscribed(clientInfo);
        if (!hasSubscribe)
        {
            throw new IllegalStateException(String.format("ClientInfo:%s should subscribe first",
                    clientInfo.toString()));
        }
    }

    private void checkStart(String destination)
    {
        if (!isStart(destination))
        {
            throw new IllegalStateException(String.format("destination:%s should start first", destination));
        }
    }


    public void setBinlogTaskFactory(BinlogTaskFactory binlogTaskFactory)
    {
        this.binlogTaskFactory = binlogTaskFactory;
    }


}
